package com.shujia.core

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo23BroadCast {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo23BroadCast")
    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    // 构建一个黑名单数据，由学生的id构成
    val blacklistRDD: RDD[String] = sc.parallelize(List("1500100018", "1500100088", "1500100188"))

    val blacklistSet: Set[String] = blacklistRDD.collect().toSet

    // 如果在算子外部定义的变量比较大时，则可以将其广播到Executor
    val blacklistSetBro: Broadcast[Set[String]] = sc.broadcast(blacklistSet)

    /**
     * RDD与RDD之间是不能嵌套的，不能在一个RDD的算子内部使用另一个RDD
     * 可以将一个数据量比较小的RDD转换成本地集合，然后再在内部使用
     */


    // 基于学生数据students.txt 再结合黑名单将 黑名单中的学生过滤出来
    val stuRDD: RDD[String] = sc.textFile("spark/data/students.txt",100)

    /**
     * 在算子内部使用算子外部的变量（基本数据类型、集合类型、自定义类、样例类）
     * 最终算子外部的变量会被封装到Task中然后发送给Executor去执行
     *
     * Task的数量由切片的数量决定
     * Executor的数量由自己根据数据量手动指定
     * 一般情况下在一个Executor中可以运行多个Task
     * 所以：Task的数量 >> Executor的数量
     *
     * 如果直接将外部变量封装到Task中，那么变量的副本会比较多
     * 优化：因为最终Task是在Executor中执行的，而且Executor的数量远远小于Task的数量
     *      所以将外部的变量通过广播的形式，将其直接发送给Executor，这样就可以大大减少变量副本的数量
     *
     */
    stuRDD.filter(line => {
      val id: String = line.split(",")(0)
      blacklistSetBro.value.contains(id)
    }).foreach(println)

  }

}
